package bx.job.httpserver;

import akka.actor.typed.ActorRef;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.javadsl.ActorContext;
import akka.http.javadsl.Http;
import akka.http.javadsl.ServerBinding;
import akka.http.javadsl.server.Route;
import bx.job.httpserver.actor.JobWorkerHeartbeatProcessorActor;
import bx.job.httpserver.actor.JobWorkerRegistProcessorActor;
import bx.job.httpserver.route.JobServerServerRoutes;
import com.typesafe.config.Config;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletionStage;

public class AkkaJobServerServer {


    public static void startJobServerServer(ActorContext<Void> actorContext) {
        ActorRef<JobWorkerRegistProcessorActor.JobWorkerRegistProcessorActorMsg> jobWorkerRegistProcessorActor =
                actorContext.spawn(JobWorkerRegistProcessorActor.create(), "bx-job-JobWorkerRegistProcessorActor");
        ActorRef<JobWorkerHeartbeatProcessorActor.JobWorkerHeartbeatProcessorActorMsg> jobClientRequestProcessorActor =
                actorContext.spawn(JobWorkerHeartbeatProcessorActor.create(), "bx-job-JobWorkerHeartbeatProcessorActor");
        JobServerServerRoutes serverServerRoutes = new JobServerServerRoutes(actorContext.getSystem(), jobWorkerRegistProcessorActor, jobClientRequestProcessorActor);
        startHttpServer(serverServerRoutes.jobServerServerRoutes(), actorContext.getSystem());
    }


    private static void startHttpServer(Route route, ActorSystem<?> system) {
        Config serverConfig = system.settings().config().getConfig("bx-job-server.server");
        String ipAddr = serverConfig.getString("ip-addr");
        int port = serverConfig.getInt("port");

        CompletionStage<ServerBinding> futureBinding = Http.get(system).newServerAt(ipAddr, port).bind(route);

        futureBinding.whenComplete((binding, exception) -> {
            if (binding != null) {
                InetSocketAddress address = binding.localAddress();
                system.log().info("bx-job-server online at http://{}:{}/", address.getHostString(), address.getPort());
            } else {
                system.log().error("Failed to bind HTTP endpoint, terminating system", exception);
                system.terminate();
            }
        });
    }
}
